package com.archstudy.springcloud.message.config;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

import java.util.Properties;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;


@Configuration
public class KafkaConfig {
    @Value("${kafka.relay.group-id}")
    private String groupId;
    @Value("${kafka.relay.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${kafka.relay.max-poll-records}")
    private int maxPollRecords;

    @Value("${kafka.relay.auto-commit}")
    private String isAutoCommit;

    @Value("${kafka.relay.auto-commit-interval}")
    private String autoCommitInterval;

    /**
     * @return
     * @Description:
     */
    @Bean("relayKafkaConsumer")
    @Scope("prototype")
    public KafkaConsumer<String, String> kafkaConsumer() {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", bootstrapServers);
        properties.setProperty("group.id", groupId);
        properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("max.poll.records", String.valueOf(maxPollRecords));
        properties.setProperty("auto.offset.reset", "earliest");
        properties.setProperty("enable.auto.commit", isAutoCommit);
        properties.setProperty("auto.commit.interval.ms", autoCommitInterval);

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        return consumer;
    }


    /**
     * @return producer
     * @Description: 业务Kafka生产者
     */
    @Bean("kafkaProducer")
    public KafkaProducer<String, String> kafkaProducer() {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", bootstrapServers);
        properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        return producer;
    }

    @Bean("relayConsumerExecutor")
    public Executor relayConsumerExecutor() {
        Executor executor = Executors.newCachedThreadPool();
        return executor;
    }


    // ... getters and setters for the properties
}
